package cn.doitedu.rtmk.demo3;

import cn.doitedu.rtmk.beans.UserEvent;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.math.BigDecimal;

public class Rule001Calculator implements RuleCalculator{

    ValueState<BigDecimal> priceSumState;
    ValueState<Integer> addCartCntState;
    Table table;

    @Override
    public void init(RuntimeContext runtimeContext) throws Exception {
        // 统计加购金额之和的状态
        priceSumState = runtimeContext.getState(new ValueStateDescriptor<BigDecimal>("price_sum", BigDecimal.class));
        // 统计加购次数的状态
        addCartCntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("add_cart_cnt", Integer.class));

        // 创建hbase连接客户端
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "doitedu:2181");
        Connection connection = ConnectionFactory.createConnection(conf);
        table = connection.getTable(TableName.valueOf("user_profile"));


    }

    @Override
    public void calculate(UserEvent userEvent, Collector<String> out) throws IOException {


        String eventId = userEvent.getEvent_id();
        long userId = userEvent.getUser_id();

        // 判断，收到的事件，是否是动态画像条件要统计的事件
        if ("add_cart".equals(eventId)) {

            /* *
             * 累加商品的价格
             */
            // 取出状态中原来的和
            BigDecimal oldSum = priceSumState.value() == null ? BigDecimal.ZERO : priceSumState.value();
            // 获取本次加购的价格
            String priceStr = userEvent.getProperties().get("price");
            // 把字符串的价格转成bigDecimal
            BigDecimal curPrice = BigDecimal.valueOf(Double.parseDouble(priceStr));
            // 原来的和 加 本次的价格
            BigDecimal added = oldSum.add(curPrice);
            // 更新到状态中
            priceSumState.update(added);


            /**
             * 累加加购的次数
             */
            // 取出状态中原来的累计值
            int oldCnt = addCartCntState.value() == null ? 0 : addCartCntState.value();
            addCartCntState.update(oldCnt + 1);


        }

        // 判断，收到的事件，是否是触发事件
        if ("order_submit".equals(eventId)) {

            // 先判断动态画像条件是否已经满足
            if (priceSumState.value() != null
                    && addCartCntState.value() != null
                    && priceSumState.value().divide(BigDecimal.valueOf(addCartCntState.value())).compareTo(BigDecimal.valueOf(100.8)) > 0) {

                // 再判断静态画像条件是否满足
                Get get = new Get((userId + "").getBytes());
                get.addColumn("f".getBytes(), "gender".getBytes());
                get.addColumn("f".getBytes(), "tag103".getBytes());

                Result result = table.get(get);
                // 从result中提取查询到的标签值
                byte[] genderBytes = result.getValue("f".getBytes(), "gender".getBytes());
                byte[] tag103Bytes = result.getValue("f".getBytes(), "tag103".getBytes());

                String gender = Bytes.toString(genderBytes);
                String tag103Str = Bytes.toString(tag103Bytes); // "25.3"
                double tag103 = Double.parseDouble(tag103Str);

                // 判断查询到的该用户的画像标签值，是否满足规则中的条件
                if (gender.equals("male") && tag103 > 10) {

                    JSONObject message = new JSONObject();
                    message.put("user_id", userId);
                    message.put("hit_time", userEvent.getAction_time());
                    message.put("rule_id", "rule-001");

                    out.collect(message.toJSONString());

                }


            }


        }



    }
}
